package com.itcast.utils

import org.apache.commons.lang3.StringUtils
import org.apache.spark.sql.Row

object RptKpiTools {
   def offLineKpi(row: Row) = {
    val rmode = row.getAs[Int]("requestmode")
    val pnode = row.getAs[Int]("processnode")
    val adRequest = if (rmode == 1 && pnode == 3) {
      List[Double](1, 1, 1)
    } else if (rmode == 1 && pnode >= 2) {
      List[Double](1, 1, 0)
    } else if (rmode == 1 && pnode >= 1) {
      List[Double](1, 0, 0)
    } else List[Double](0, 0, 0)
    val iseffective = row.getAs[Int]("iseffective")
    val isbilling = row.getAs[Int]("isbilling")
    val isbid = row.getAs[Int]("isbid")
    val iswin = row.getAs[Int]("iswin")
    val adorderid = row.getAs[Int]("adorderid")
    val adRtb = if (iseffective == 1 && isbilling == 1 && isbid == 1 && adorderid != 0) List[Double](1, 0)
    else if (iseffective == 1 && isbilling == 1 && iswin == 1) List[Double](0, 1)
    else List[Double](0, 0)
    val adShowClick = if (rmode == 2 && iseffective == 1) List[Double](1, 0)
    else if (rmode == 3 && iseffective == 1) List[Double](0, 1)
    else List[Double](0, 0)
    val winPrice = row.getAs[Double]("winprice")
    val adpayment = row.getAs[Double]("adpayment")
    val adCost = if (iseffective == 1 && isbilling == 1 && iswin == 1) List[Double](winPrice / 1000.0, adpayment / 1000)
    else List[Double](0, 0)


     adRequest ++ adRtb ++ adShowClick ++ adCost
  }

  val notEmptyAppName =(appId:String,appName:String) =>{
    if (StringUtils.isEmpty(appName)){
      val jedis = Jpools.getJedis
      val appName = jedis.hget("appdict",appId)
      jedis.close()
      appName
    }else appName
  }


}
